package com.mitake.task;

import java.io.FileOutputStream;
import java.util.Arrays;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import com.mitake.network.HttpData;
import com.mitake.network.ISocketListener;
import com.mitake.network.MitakeHttpParams;
import com.mitake.network.MitakeSocket;
import com.mitake.parse.MitakeQuote;
import com.mitake.parse.Stock;
import com.mitake.utils.DateUtility;
import com.mitake.utils.MitakeLog;

public class QuoteCompareTask
{
	private static final String COLUMN_SEPARATOR = String.valueOf((char)0x02);

	private static boolean DEBUG = true;
	private static int TIMEOUT = 1000 * 1000000;
	private static QuoteCompareTask instance;
	/**
     * 封包序号
     */
    private AtomicInteger packageNo;

	private int hostCount;
	private String[] stock;
	private MitakeSocket[] socket;
	private Hashtable<String, TelegramGroup> queue;
	private ConcurrentLinkedQueue<TelegramGroup> saveTelegramQueue;
	private long runInterval;
	private boolean running;
	private String logPath;
	/**
	 * 发送比数
	 */
	private long writeTelegramCount;
	/**
	 * 回传笔数
	 */
	private long readTelegramCount;
	
	private QuoteCompareTask() {}
	
	
	public static synchronized QuoteCompareTask getInstance()
	{
		if(instance == null)
		{
			instance = new QuoteCompareTask();
		}
		
		return instance;
	}
	
	public void setRunInterval(long runInterval)
	{
		this.runInterval = runInterval;
	}
	
	public void setStockList(String[] stock)
	{
		this.stock = stock;
	}
	
	public void setLogPath(String logPath)
	{
		this.logPath = logPath;
	}

	public boolean isRunning()
	{
		return running;
	}
	
	public String[] getTelegramCount()
	{
		return new String[]{String.valueOf(writeTelegramCount), String.valueOf(readTelegramCount)};
	}
	
	public String[][] getAllSocketStatus()
	{
		String[][] data = null;
		
		if(socket != null)
		{
			data = new String[hostCount][2];
			
			for(int i = 0 ; i < socket.length ; i++)
			{
				data[i][0] = socket[i].desc + "_" + socket[i].ip + ":" + socket[i].port;
				data[i][1] = socket[i].isConnected() ? "0" : "1";
				
//				if(DEBUG)
//				{
//					MitakeLog.log("连线状态==" + socket[i].ip + ":" + socket[i].port + "_" + socket[i].isConnected());
//				}
			}
		}
		
		return data;
	}
	
	private class SendAPI implements Runnable
	{
		public void run()
		{
			long beforeTime = 0;
			/**
			 * 組股票代碼
			 */
			StringBuilder stocks = new StringBuilder();
			
			for(int i = 0 ; i < stock.length ; i++)
			{
				stocks.append(stock[i]).append(",");
			}
			
			stocks.setLength(stocks.length() - 1);
			
			String list = stocks.toString();
			
			String pn = null;
			
			while(running)
			{
				if(System.currentTimeMillis() - beforeTime > runInterval)
				{
					pn = String.valueOf(packageNo.incrementAndGet());
					
					for(int i = 0 ; i < hostCount ; i++)
					{
						/**
						 * 組成電文
						 */
						MitakeHttpParams params = new MitakeHttpParams();
						params.api = "/v2/quotentrd1";
//						params.api = "/v2/quote";
						params.C2SDataType = MitakeHttpParams.C_S_DATA_TYPE_STRING;
						params.S2CDataType = MitakeHttpParams.S_C_DATA_TYPE_BYTES;
						params.headers = new String[][] {{"Token", "MitakeWeb"}, {"Symbol", list}};
						params.packageNo = pn;
						params.method = "GET";

						if(socket[i].write(params))
						{
							writeTelegramCount++;
						}
					}
					
					beforeTime = System.currentTimeMillis();
				}
				
				try
				{
					TimeUnit.MILLISECONDS.sleep(10);
				}
				catch(InterruptedException e)
				{
					e.printStackTrace();
				}
			}
		}
	}
	
	public void start(String[][] host)
	{
		hostCount = host.length;
		/**
		 * 初始封包序號
		 */
		packageNo = new AtomicInteger();
		
		socket = new MitakeSocket[hostCount];
		
		queue = new Hashtable<String, TelegramGroup>();
		
		saveTelegramQueue = new ConcurrentLinkedQueue<TelegramGroup>();
		/**
		 * 建立所有機房連線
		 */
		for(int i = 0 ; i < host.length ; i++)
		{
			socket[i] = new MitakeSocket(host[i], new ISocketListener()
			{
				@Override
				public void onNetworkStatusChanged(MitakeSocket mitakeSocket, String packageNo, String code)
				{
					if(code.equals("00"))
                    {
                        /**
                         * 连线成功
                         */
						if(DEBUG)
						{
							MitakeLog.log("连线成功==" + mitakeSocket.ip + ":" + mitakeSocket.port + "==" + packageNo);
						}
                    }
                    else if(code.equals("01"))
                    {
                        /**
                         * 断线,无需自动连线
                         */
                    	if(DEBUG)
                    	{
                    		MitakeLog.log("断线,无需自动连线==" + mitakeSocket.ip + ":" + mitakeSocket.port);
                    	}
                    }
                    else if(code.equals("02"))
                    {
                        /**
                         * 無預期斷線,主動重新連線
                         */
                    	if(DEBUG)
                    	{
                    		MitakeLog.log("無預期斷線==" + mitakeSocket.ip + ":" + mitakeSocket.port);
                    	}
                    	
                    	try
						{
							TimeUnit.MILLISECONDS.sleep(100);
						}
						catch(InterruptedException e)
						{
							e.printStackTrace();
						}
                    	
                    	mitakeSocket.start();
                    }
				}

				@Override
				public void onContent(HttpData httpData)
				{
					readTelegramCount++;
					HashMap<String, String> header = httpData.headers;
                    String packageNo = header.get("k");
                    /**
                     * 找不到封包序號就丟掉
                     */
                    if(packageNo != null)
                    {
                    	/**
                    	 * 逾時的封包丟掉
                    	 */
                    	if(httpData.responseTime - httpData.sendTime <= TIMEOUT)
                    	{
	                    	TelegramGroup telegramGroup = queue.get(packageNo);
	
	                        if(telegramGroup == null)
	                        {
	                        	telegramGroup = new TelegramGroup(hostCount);
	                        	telegramGroup.count = 0;
	                        	telegramGroup.packageNo = packageNo;
	                        	telegramGroup.initTime = System.nanoTime();
	                        	telegramGroup.stock = stock;
	                        	
	                        	queue.put(packageNo, telegramGroup);
	                        }

	                        telegramGroup.httpData[telegramGroup.count] = httpData;
	                    	telegramGroup.count++;
                    	}
                    }
				}
			});
			
			socket[i].start();
		}
		/**
		 * 比较机制启动
		 */
		running = true;
		new Thread(new SendAPI()).start();
		new Thread(new CompareTask()).start();
		new Thread(new SaveTask()).start();
	}
	
	public void stop()
	{
		running = false;
		
		if(socket != null)
		{
			for(int i = 0 ; i < socket.length ; i++)
			{
				if(socket[i] != null)
				{
					socket[i].disconnect("", false);
				}
			}
		}

		packageNo = null;
		
		queue.clear();
		queue = null;
	}
	
	/**
	 * 比对数据
	 */
	private class CompareTask implements Runnable
	{
		public void run()
		{
			while(running)
			{
				if(queue != null && !queue.isEmpty())
				{
					for(Enumeration<TelegramGroup> e = queue.elements() ; e.hasMoreElements() ; )
					{
						TelegramGroup telegramGroup = e.nextElement();
						
						if(telegramGroup.count >= hostCount)
						{
							/**
							 * 進行內容比對
							 */
							for(int i = 0, len = telegramGroup.count - 1 ; i < len ; i++)
							{
								/**
								 * 內容不相等才進行剖析
								 */
								if(!Arrays.equals(telegramGroup.httpData[i].b, telegramGroup.httpData[i + 1].b))
								{
									telegramGroup.hasDiff = true;
									break;
								}
							}
							/**
							 * 放入欲储存的queue
							 */
							saveTelegramQueue.offer(telegramGroup);

							queue.remove(telegramGroup.packageNo);
						}
						else if(System.currentTimeMillis() - telegramGroup.initTime > TIMEOUT)
						{
							/**
							 * 超過規定時間沒有返回,就當有問題
							 */
							telegramGroup.hasDiff = true;
							/**
							 * 放入欲储存的queue
							 */
							saveTelegramQueue.offer(telegramGroup);
							
							queue.remove(telegramGroup.packageNo);
						}
					}
				}
				
				try
				{
					TimeUnit.MILLISECONDS.sleep(100);
				}
				catch(InterruptedException e)
				{
					e.printStackTrace();
				}
			}
		}
	}
	
	private class SaveTask implements Runnable
	{
		public void run()
		{
			StringBuilder data = new StringBuilder();
			
			while(running)
			{
				TelegramGroup telegramGroup = saveTelegramQueue.poll();

				if(telegramGroup == null)
				{
					try
					{
						TimeUnit.MILLISECONDS.sleep(10);
					}
					catch(Exception e)
					{
						e.printStackTrace();
					}
				}
				else if(telegramGroup.count > 0)
				{
					data.setLength(0);
					
					Stock[] stock = new Stock[telegramGroup.count];
					/**
					 * 开启剖析电文并存入档案内
					 */
					for(int i = 0 ; i < telegramGroup.count ; i++)
					{
						try
						{
							stock[i] = MitakeQuote.parseDetail(new String(telegramGroup.httpData[i].b, "utf-8"));
							
							stock[i].host = telegramGroup.httpData[i].host;
							stock[i].name = telegramGroup.httpData[i].name;
							stock[i].writeTime = String.valueOf(telegramGroup.httpData[i].sendTime);
							stock[i].readTime = String.valueOf(telegramGroup.httpData[i].responseTime);
							stock[i].responseTime = String.valueOf((telegramGroup.httpData[i].responseTime - telegramGroup.httpData[i].sendTime) / 1000000) + "ms";

							appendData(data, stock[i]);
						}
						catch(Exception e)
						{
							e.printStackTrace();
						}
					}
					
					try
					{
						data.append("----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------\r\n");
						
						FileOutputStream fos = new FileOutputStream(logPath + "quote-" + DateUtility.getNowDate() + ".log", true);
						fos.write(data.toString().getBytes("utf-8"));
						fos.flush();
						fos.close();
						/**
						 * 有差异的数据放到另一个档案
						 */
						if(telegramGroup.hasDiff)
						{
							Stock[] list = new Stock[stock.length + 1];
							list[0] = new Stock();
							System.arraycopy(stock, 0, list, 1, stock.length);
							CompareProcess.compare(list);
							
							FileOutputStream fos1 = new FileOutputStream(logPath + "quote-diff-" + DateUtility.getNowDate() + ".log", true);
							fos1.write(data.toString().getBytes("utf-8"));
							fos1.flush();
							fos1.close();
						}
					}
					catch(Exception e)
					{
						e.printStackTrace();
					}
				}
				
				try
				{
					TimeUnit.MILLISECONDS.sleep(100);
				}
				catch(InterruptedException e)
				{
					e.printStackTrace();
				}
			}
		}
	}
	
	private void appendData(StringBuilder data, Stock stock)
	{
		data.append(stock.host);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.writeTime);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.readTime);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.responseTime);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.datetime.substring(0, 8));
		data.append(COLUMN_SEPARATOR);
		data.append(stock.datetime.substring(8));
		data.append(COLUMN_SEPARATOR);
		data.append(stock.id);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.name);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.status);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.lastPrice);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.highPrice);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.lowPrice);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.openPrice);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.preClosePrice);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.changeRate);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.volume);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.nowVolume);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.turnoverRate);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.limitUP);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.limitDown);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.averageValue);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.change);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.amount);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.volumeRatio);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.buyPrice);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.sellPrice);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.buyVolume);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.sellVolume);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.totalValue);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.flowValue);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.netAsset);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.pe);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.roe);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.capitalization);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.circulatingShares);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.buyPricesTmp);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.buyVolumesTmp);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.buySingleVolumesTmp);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.sellPricesTmp);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.sellVolumesTmp);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.sellSingleVolumesTmp);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.amplitudeRate);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.receipts);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.tradeTickTmp);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.upCount);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.sameCount);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.downCount);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.contractID);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.objectID);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.stockSymble);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.stockType);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.stockUnit);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.exePrice);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.startDate);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.endDate);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.exeDate);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.delDate);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.expDate);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.version);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.presetPrice);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.setPrice);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.stockClose);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.stockLast);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.isLimit);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.marginUnit);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.roundLot);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.inVal);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.timeVal);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.preInterest);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.openInterest);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.tradePhase);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.remainDate);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.leverage);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.premium);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.orderRatio);
		data.append(COLUMN_SEPARATOR);
		data.append(stock.ip);
		data.append("\r\n");
	}
	
	private void setFileTitle(StringBuilder data)
	{
		data.append("机房:");
		data.append(COLUMN_SEPARATOR);
		data.append("发送时间:");
		data.append(COLUMN_SEPARATOR);
		data.append("回传时间:");
		data.append(COLUMN_SEPARATOR);
		data.append("响应时间:");
		data.append(COLUMN_SEPARATOR);
		data.append("报价日期:");
		data.append(COLUMN_SEPARATOR);
		data.append("报价时间:");
		data.append(COLUMN_SEPARATOR);
		data.append("股票代码:");
		data.append(COLUMN_SEPARATOR);
		data.append("名称:");
		data.append(COLUMN_SEPARATOR);
		data.append("股票状态:");
		data.append(COLUMN_SEPARATOR);
		data.append("最新价:");
		data.append(COLUMN_SEPARATOR);
		data.append("最高价:");
		data.append(COLUMN_SEPARATOR);
		data.append("最低价:");
		data.append(COLUMN_SEPARATOR);
		data.append("开盘价:");
		data.append(COLUMN_SEPARATOR);
		data.append("昨收价:");
		data.append(COLUMN_SEPARATOR);
		data.append("涨幅:");
		data.append(COLUMN_SEPARATOR);
		data.append("总量:");
		data.append(COLUMN_SEPARATOR);
		data.append("成交量:");
		data.append(COLUMN_SEPARATOR);
		data.append("换手率:");
		data.append(COLUMN_SEPARATOR);
		data.append("涨停价:");
		data.append(COLUMN_SEPARATOR);
		data.append("跌停价:");
		data.append(COLUMN_SEPARATOR);
		data.append("均价:");
		data.append(COLUMN_SEPARATOR);
		data.append("涨跌:");
		data.append(COLUMN_SEPARATOR);
		data.append("成交金额:");
		data.append(COLUMN_SEPARATOR);
		data.append("量比:");
		data.append(COLUMN_SEPARATOR);
		data.append("买一价:");
		data.append(COLUMN_SEPARATOR);
		data.append("卖一价:");
		data.append(COLUMN_SEPARATOR);
		data.append("外盘量:");
		data.append(COLUMN_SEPARATOR);
		data.append("内盘量:");
		data.append(COLUMN_SEPARATOR);
		data.append("总值:");
		data.append(COLUMN_SEPARATOR);
		data.append("流值:");
		data.append(COLUMN_SEPARATOR);
		data.append("净资产:");
		data.append(COLUMN_SEPARATOR);
		data.append("PE(市盈):");
		data.append(COLUMN_SEPARATOR);
		data.append("ROE(净资产收益率):");
		data.append(COLUMN_SEPARATOR);
		data.append("总股本:");
		data.append(COLUMN_SEPARATOR);
		data.append("流通股:");
		data.append(COLUMN_SEPARATOR);
		data.append("买十档价:");
		data.append(COLUMN_SEPARATOR);
		data.append("买十档量:");
		data.append(COLUMN_SEPARATOR);
		data.append("买十档量(单量):");
		data.append(COLUMN_SEPARATOR);
		data.append("卖十档价:");
		data.append(COLUMN_SEPARATOR);
		data.append("卖十档量:");
		data.append(COLUMN_SEPARATOR);
		data.append("卖十档量（单量）:");
		data.append(COLUMN_SEPARATOR);
		data.append("振幅比率:");
		data.append(COLUMN_SEPARATOR);
		data.append("收益:");
		data.append(COLUMN_SEPARATOR);
		data.append("成交明细:");
		data.append(COLUMN_SEPARATOR);
		data.append("上涨家数:");
		data.append(COLUMN_SEPARATOR);
		data.append("平盘家数:");
		data.append(COLUMN_SEPARATOR);
		data.append("下跌家数:");
		data.append(COLUMN_SEPARATOR);
		data.append("合约代码:");
		data.append(COLUMN_SEPARATOR);
		data.append("标的证券代码:");
		data.append(COLUMN_SEPARATOR);
		data.append("标的证券简称:");
		data.append(COLUMN_SEPARATOR);
		data.append("标的证券类型:");
		data.append(COLUMN_SEPARATOR);
		data.append("合约单位:");
		data.append(COLUMN_SEPARATOR);
		data.append("执行价格:");
		data.append(COLUMN_SEPARATOR);
		data.append("首交易日:");
		data.append(COLUMN_SEPARATOR);
		data.append("最后交易日:");
		data.append(COLUMN_SEPARATOR);
		data.append("行权日:");
		data.append(COLUMN_SEPARATOR);
		data.append("交割日:");
		data.append(COLUMN_SEPARATOR);
		data.append("到期日:");
		data.append(COLUMN_SEPARATOR);
		data.append("合约版号:");
		data.append(COLUMN_SEPARATOR);
		data.append("前结算价:");
		data.append(COLUMN_SEPARATOR);
		data.append("当日结算价:");
		data.append(COLUMN_SEPARATOR);
		data.append("标的证券昨收:");
		data.append(COLUMN_SEPARATOR);
		data.append("标的证券价格:");
		data.append(COLUMN_SEPARATOR);
		data.append("有无涨跌限制:");
		data.append(COLUMN_SEPARATOR);
		data.append("合约保证金:");
		data.append(COLUMN_SEPARATOR);
		data.append("一手的合约数:");
		data.append(COLUMN_SEPARATOR);
		data.append("内涵价值:");
		data.append(COLUMN_SEPARATOR);
		data.append("时间价值:");
		data.append(COLUMN_SEPARATOR);
		data.append("昨日持仓量(张):");
		data.append(COLUMN_SEPARATOR);
		data.append("持仓量:");
		data.append(COLUMN_SEPARATOR);
		data.append("交易时段:");
		data.append(COLUMN_SEPARATOR);
		data.append("剩余天数:");
		data.append(COLUMN_SEPARATOR);
		data.append("杠杆比率:");
		data.append(COLUMN_SEPARATOR);
		data.append("溢价率:");
		data.append(COLUMN_SEPARATOR);
		data.append("委比:");
		data.append(COLUMN_SEPARATOR);
		data.append("ip: ");
		data.append("\r\n");
	}
}
